/*
 * Copyright (C) 2006 Moritz Tuerk <brotkasten@users.sourceforge.net>
 * 
 * This program is free software; you can redistribute it and/or modify it under the terms 
 * of the GNU General Public License as published by the Free Software Foundation; either 
 * version 2 of the License, or (at your option) any later version.
 * 
 * This program is distributed in the hope that it will be useful, 
 * but WITHOUT ANY WARRANTY; without even the implied 
 * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. 
 * See the GNU General Public License for more details.
 *  
 *  You should have received a copy of the GNU General Public License along with 
 *  this program; if not, write to the Free Software Foundation, 
 *  Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110, USA
 */
package mo.network.server;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketException;
import java.util.Hashtable;

import mo.common.util.Heap;
import mo.network.client.listeners.IReceiver;
import mo.network.client.listeners.ISender;
import mo.network.client.listeners.ReceiverEventQueue;
import mo.network.client.listeners.SenderEvent;
import mo.network.client.listeners.SenderEventQueue;
import mo.network.server.data.Fragment;
import mo.network.server.data.Packet;

final class ServerThread extends Thread {
	/**
	 * saves the acknoledge message to be sent to the client if a fragment was
	 * received successfully. The client also has to answer with this message
	 * to
	 * accept a fragment from the server.
	 */
	public final static String FRAGMENT_ACK_MSG = "ACK#";

	/**
	 * used to lock the wait staus
	 */
	private final static Object LOCK = new Object();

	/**
	 * saves the message send by the server if a packet failed to send. This
	 * message must be answered by the client or no further packets are send
	 * to
	 * the client!
	 */
	public final static String PACKET_FAIL_MSG = "PF#";

	/**
	 * saves the stream for incoming traffic from the client
	 */
	private final InputStream clientIn;

	/**
	 * saves the stream for outgoing traffic to the client
	 */
	private final OutputStream clientOut;

	/**
	 * saves the socket to read and write from and to the client
	 */
	private final Socket clientSock;

	/**
	 * saves the packets that are currently being recieved by the server. The
	 * table saves the packets by their packet id.
	 */
	private final Hashtable<Long, Packet> currentDownloads;

	/**
	 * saves the packet stack number. Each packet gets a unique number to
	 * identify the packet sent by the server.
	 */
	long packetStack;

	/**
	 * Eventqueue for reporting received {@link Packet} objects to the
	 * {@linkplain #addReceiveListener(IReceiver) registered} listeners.
	 */
	private final ReceiverEventQueue receiverEventQueue;

	private final SenderEventQueue senderQueue;

	/**
	 * saves the server instance that started this thread
	 */
	private final NetworkServer server;

	/**
	 * saves the flag indicating if this thread is still running
	 */
	boolean stillAlive;

	/**
	 * the queue for uploading packets to the client.
	 */
	private final Heap<Packet> uploadQueue;

	/**
	 * saves the time when the wait phase started
	 */
	private long waitStart;

	/**
	 * this is used to make this thread wait.<br>
	 * 0 means don't wait<br>
	 * 1 means wait<br>
	 * 2 means thread is waiting<br>
	 * 3 means continue and quit waiting<br>
	 */
	private int waitStatus;

	/**
	 * This creates a new Server Thread
	 * 
	 * @param server
	 *            saves the server that started the thread
	 * @param clientSock
	 *            saves the socket to communicate to the client
	 * @throws IOException
	 *             if the streams couldn't be read from the socket
	 */
	public ServerThread(final NetworkServer server, final Socket clientSock)
			throws IOException {
		super("ServerThread");
		this.server = server;
		this.clientSock = clientSock;
		this.clientIn = clientSock.getInputStream();
		this.clientOut = clientSock.getOutputStream();

		this.senderQueue = new SenderEventQueue();
		this.uploadQueue = new Heap<Packet>();
		this.currentDownloads = new Hashtable<Long, Packet>();
		this.receiverEventQueue = new ReceiverEventQueue();

		this.packetStack = 0;
		this.waitStatus = 0;
		this.stillAlive = false;
	}

	/**
	 * Adds a new listener for incoming packets to this server thread. A
	 * listener cannot be registered twice!
	 * 
	 * @param receiver
	 *            the listener to add to the list of receive listeners
	 */
	public void addReceiveListener(final IReceiver receiver) {
		this.receiverEventQueue.addIReceiver(receiver);
	}

	/**
	 * Adds a new listener for outgoing packets to this server thread. A
	 * listener cannot be registered twice!
	 * 
	 * @param sender
	 *            the listener to add to the list of receive listeners
	 */
	public void addSendListener(final ISender sender) {
		this.senderQueue.addISender(sender);
	}

	/**
	 * This method is called to create an acknowledge message for the client,
	 * that a fragment was sent successfully.
	 * 
	 * @param packetNo
	 *            saves the packet number to which the fragment belongs.
	 * @param fragmentNo
	 *            saves the fragments number
	 * @param clientSock
	 *            saves the socket to create the ack message for
	 * @return an acknowledge packet for the given client (socket)
	 */
	private Packet createAckMsg(final long packetNo, final int fragmentNo,
			final Socket clientSock) {
		final StringBuffer pkgAckMsgBuf =
				new StringBuffer(ServerThread.FRAGMENT_ACK_MSG);
		pkgAckMsgBuf.append(packetNo);
		pkgAckMsgBuf.append("#");
		pkgAckMsgBuf.append(fragmentNo);

		final Packet result =
				new Packet(pkgAckMsgBuf.toString().getBytes(), clientSock);

		result.setType(Packet.PACKET_TYPE_ACK);
		result.setPriority(Packet.PACKET_PRIORITY_MAX - 1);

		return result;
	}

	/**
	 * This method is called to create a packet fail message which is send to
	 * the client when a packet could not be sent. This is important to keep
	 * the
	 * internal packet tables clean. Fail packets remain in the first place in
	 * the queue until they are answered by the client, so they migth possibly
	 * block the queue until it is answered by the client!
	 * 
	 * @param packetNo
	 *            saves the id of the packet that could not be sent
	 * @param clientSock
	 *            saves the client socket to send the packet to
	 * @return a packet with a packet fail message
	 */
	private Packet createFailMsg(final long packetNo, final Socket clientSock) {
		final StringBuffer pkgFailMsgBuf =
				new StringBuffer(ServerThread.PACKET_FAIL_MSG);
		pkgFailMsgBuf.append(packetNo);

		final Packet result =
				new Packet(pkgFailMsgBuf.toString().getBytes(), clientSock);
		result.setType(Packet.PACKET_TYPE_PF);
		result.setPriority(0);

		return result;
	}

	/**
	 * This method is called do docompose an acknowledgement message.
	 * 
	 * @param ackMsg
	 *            saves the message to be tested.
	 * @return a string array of lenght 3 or <code>null</code>. If
	 *         <code>null</code> is returned, then the given message is no
	 *         acknowledge message, else the string array contains the
	 *         following:<br>
	 *         <ul>
	 *         <li>[0] = "ACK"</li>
	 *         <li>[1] = packet id (as string)</li>
	 *         <li>[2] = fragment id (as string)</li>
	 *         </ul>
	 */
	private String[] decomposeAckMessage(final String ackMsg) {
		final String result[] = ackMsg.split("#");

		// msg is like 'ACK#PKGNO#FRAGNO'
		if (result.length != 3) {
			return null;
		}

		// is it an ack message ?
		if (ServerThread.FRAGMENT_ACK_MSG.startsWith(result[0])) {
			return result;
		}

		return null;
	}

	/**
	 * This method is called do docompose an packet fail message.
	 * 
	 * @param pfMsg
	 *            saves the message to be tested.
	 * @return a string array of lenght 2 or <code>null</code>. If
	 *         <code>null</code> is returned, then the given message is no
	 *         packet fail message, else the string array contains the
	 *         following:<br>
	 *         <ul>
	 *         <li>[0] = "PF"</li>
	 *         <li>[1] = packet id (as string)</li>
	 *         </ul>
	 */
	private String[] decomposePFMessage(final String pfMsg) {
		final String result[] = pfMsg.split("#");

		// msg is like 'FAIL#PKGNO'
		if (result.length != 2) {
			return null;
		}

		// is it an ack message ?
		if (ServerThread.PACKET_FAIL_MSG.startsWith(result[0])) {
			return result;
		}

		return null;
	}

	/**
	 * This method can be called from the server to make this thread go on and
	 * quit waiting
	 */
	void doContinue() {
		synchronized (ServerThread.LOCK) {
			this.waitStatus = 3;
		}
	}

	/**
	 * This method can be called from the server to make this thread wait.
	 */
	void doWait() {
		synchronized (ServerThread.LOCK) {
			this.waitStatus = 1;
		}
		while (this.waitStatus != 2) {
			try {
				Thread.sleep(100);
			} catch (final InterruptedException ire) {
				// do nothing
			}
		}
	}

	/**
	 * This method is called everytime a packet couldn't be sent (because of a
	 * timeout or an exception)
	 * 
	 * @param failed
	 *            the packet that was not sent.
	 */
	void firePacketFailed(final Packet failed) {
		this.senderQueue.enqueue(failed, SenderEvent.PACKET_FAILED);
	}

	/**
	 * This method is called by the server threads to indicate that a packet
	 * was
	 * received from a client.
	 * 
	 * @param thePacket
	 *            saves the packet that was received.
	 */
	void firePacketReceived(final Packet thePacket) {
		this.receiverEventQueue.enqueue(thePacket);
	}

	/**
	 * This method is called by the server threads to indicate that a packet
	 * was
	 * successfully sent to a client.
	 * 
	 * @param sent
	 *            saves the packet that was sent
	 */
	protected void firePacketSent(final Packet sent) {
		this.senderQueue.enqueue(sent, SenderEvent.PACKET_SENT);
	}

	/**
	 * This method is called internally to get the socket for the
	 * communication
	 * to the client
	 * 
	 * @return the client socket for communication
	 */
	Socket getClientSocket() {
		return this.clientSock;
	}

	/**
	 * This method is called by the main loop of the server thread to handle
	 * all
	 * downloads from the client. It checks if there is data available from
	 * the
	 * client.
	 * 
	 * @return <code>true</code> if an download was performed, else
	 *         <code>false</code>
	 * @throws SocketException
	 * @throws IOException
	 */
	private boolean handleDownloads() throws SocketException, IOException {
		boolean actionTaken = false;

		if (this.clientIn.available() > 0
				&& NetworkServer.reserveDownload(Fragment.MAX_PAYLOAD
						+ Fragment.OVERHEAD)) {

			actionTaken = true;

			final byte lenBuf[] = new byte[8];

			int readCount = this.clientIn.read(lenBuf);
			while (readCount < lenBuf.length) {
				readCount +=
						this.clientIn.read(lenBuf, readCount, lenBuf.length
								- readCount);
			}

			final int length = (int) Long.parseLong(new String(lenBuf), 16);

			final byte buffer[] = new byte[length - lenBuf.length];

			readCount = this.clientIn.read(buffer);
			while (readCount < buffer.length) {
				readCount +=
						this.clientIn.read(buffer, readCount, buffer.length
								- readCount);
			}

			// release unused bandwidth!
			final int toRelease =
					Fragment.MAX_PAYLOAD + Fragment.OVERHEAD
							- (buffer.length + lenBuf.length);
			NetworkServer.releaseDownload(toRelease);

			final byte fragment[] = new byte[lenBuf.length + buffer.length];
			System.arraycopy(lenBuf, 0, fragment, 0, lenBuf.length);
			System.arraycopy(buffer, 0, fragment, lenBuf.length, buffer.length);

			final Fragment nextFragment = Fragment.deserialize(fragment);

			// get packet id and check if we are currently
			// downloading this packet...
			final Long pkgId = nextFragment.getPacketId();
			Packet current = this.currentDownloads.get(pkgId);

			boolean sendAckMsg = true;

			switch (nextFragment.getType()) {
			case Packet.PACKET_TYPE_ACK:
				// decompose the ack message
				final String ackDecomp[] =
						this.decomposeAckMessage(new String(nextFragment
								.getPayload()));

				sendAckMsg = false;

				synchronized (ServerThread.LOCK) {
					final int size = this.uploadQueue.size();
					if (size > 0) {
						final int packetId = Integer.parseInt(ackDecomp[1]);

						final Comparable<Packet>[] elements =
								this.uploadQueue.getElements();
						Packet packet = null;

						for (int i = 0; i < size && packet == null; i++) {
							if (((Packet) elements[i]).getNo() == packetId) {
								packet = (Packet) elements[i];
							}
						}

						if (packet != null) {
							packet.markAsSent(Integer.parseInt(ackDecomp[2]));
						}
					}
				}

				break;

			case Packet.PACKET_TYPE_PF:
				// decompose the fail message
				final String pfDecomp[] =
						this.decomposePFMessage(new String(nextFragment
								.getPayload()));

				final Long failID = Long.parseLong(pfDecomp[1]);
				this.currentDownloads.remove(failID);

				break;

			default:
				// if no ack message, check current downloaded packet
				if (current == null) {
					current = new Packet(this.clientSock);
					current.setNo(pkgId);

					this.currentDownloads.put(pkgId, current);

				}

				current.addFragment(nextFragment);
			}

			// the packet is complete ?
			if (current != null && current.isComplete()) {
				if (sendAckMsg) {
					this.firePacketReceived(current);
					this.server.fireGlobalPacketReceived(current);
				}
				this.currentDownloads.remove(pkgId);
			}

			if (sendAckMsg) {
				this.queue(this.createAckMsg(nextFragment.getPacketId(),
						nextFragment.getNumber(), this.clientSock));
			}
		}

		return actionTaken;
	}

	/**
	 * This method is called by the main loop of the server thread to handle
	 * all
	 * uploads to the client. It checks the packet queue for this thread for
	 * outgoing packets and sends them to the client.
	 * 
	 * @return <code>true</code> if an upload was performed, else
	 *         <code>false</code>
	 * @throws SocketException
	 * @throws IOException
	 */
	private boolean handleUploads() throws SocketException, IOException {
		Packet next = null;
		boolean actionTaken = false;

		synchronized (ServerThread.LOCK) {
			if (this.uploadQueue.size() > 0) {
				next = this.uploadQueue.deleteFirst();
			}
		}

		if (next != null) {
			if (next.isPacketFailed()) {
				if (next.getType() == Packet.PACKET_TYPE_PF) {
					final String[] pfDecomp =
							this.decomposePFMessage(new String(next
									.getPayLoad()));

					final Packet retry =
							this.createFailMsg(Long.parseLong(pfDecomp[1]),
									next.getSocket());
					this.queue(retry);

				} else {
					this.queue(this.createFailMsg(next.getNo(), next
							.getSocket()));

					this.firePacketFailed(next);
					this.server.fireGlobalPacketFailed(next);
				}

			} else if (next.isComplete()) {
				if (next.getType() != Packet.PACKET_TYPE_PING
						&& next.getType() != Packet.PACKET_TYPE_PF) {
					this.firePacketSent(next);
					this.server.fireGlobalPacketSent(next);
				}

			} else {
				final long now = System.currentTimeMillis();
				Fragment next2send = next.getNext2Send(false, now);
				boolean addBack = true;

				int reservedSize = 0;
				if (next2send != null
						&& NetworkServer.reserveUpload((reservedSize =
								next2send.getSize()))) {

					next2send = next.getNext2Send(true, now);
					if (next2send != null) {
						actionTaken = true;
						this.clientOut.write(next2send.serialize());
						this.clientOut.flush();

						if (next.getType() == Packet.PACKET_TYPE_ACK) {
							addBack = false;
						}
					} else {
						NetworkServer.releaseUpload(reservedSize);
					}
				}
				if (addBack) {
					synchronized (ServerThread.LOCK) {
						this.uploadQueue.addElement(next);
					}
				}
			}
		}

		return actionTaken;
	}

	/**
	 * This is called by the network server to add a new packet for uploading
	 * to
	 * the client
	 * 
	 * @param newPacket
	 *            saves the packet to send
	 * @param prioPacket
	 *            can be set to <code>true</code> if the queued packet is an
	 *            acknowledge or ping packet. This will queue the packet as
	 *            first packet in the queue
	 */
	void queue(final Packet newPacket) {
		synchronized (ServerThread.LOCK) {
			newPacket.setNo(this.packetStack++);
			this.uploadQueue.addElement(newPacket);
		}
	}

	/**
	 * Removes a receive listener from the list of listeners.
	 * 
	 * @param receiver
	 *            the receive listener to be removed from the list
	 */
	public void removeReceiveListener(final IReceiver receiver) {
		this.receiverEventQueue.removeIReceiver(receiver);
	}

	/**
	 * Removes a send listener from the list of listeners.
	 * 
	 * @param sender
	 *            the send listener to be removed from the list
	 */
	public void removeSendListener(final ISender sender) {
		this.senderQueue.removeISender(sender);
	}

	/**
	 * the main loop of the server thread. It handles the up and downloads for
	 * a
	 * connected client
	 */
	@Override
	public void run() {
		if (this.stillAlive) {
			System.err.println("YOU CANNOT RUN ME TWICE!");
			return;
		}

		this.stillAlive = true;
		while (this.stillAlive && this.server.stillAlive) {
			if (this.waitStatus == 0) {
				boolean download = false;
				boolean upload = false;

				try {
					upload = this.handleUploads();
					download = this.handleDownloads();

					// no action meas wait !
					if (!upload && !download) {
						try {
							Thread.sleep(5);
						} catch (final InterruptedException ire) {
							// nothing to do...
						}
					}

				} catch (final Exception e) {
					this.stillAlive = false;

					System.err.println("[OH OH ... ServerThread blew up!]");
					e.printStackTrace();

					this.server.disconnectClient(this.clientSock);
				}

			} else {
				switch (this.waitStatus) {
				case 1:
					synchronized (ServerThread.LOCK) {
						this.waitStatus = 2;
						this.waitStart = System.currentTimeMillis();
					}
					break;

				case 3:
					synchronized (ServerThread.LOCK) {
						this.waitStatus = 0;
						final long waitTime =
								System.currentTimeMillis() - this.waitStart;
						this.updatePacketTime(waitTime);
					}
					break;

				default:
					// try {
					// Thread.sleep(50);
					// } catch (final InterruptedException ire) {
					// // nothing to do...
					// }
				}
			}
		}

		// when stopping ... try disconnect
		try {
			this.clientSock.close();
		} catch (final Exception e) {

		}
	}

	/**
	 * This method is called when a wait phase is over. This updates all
	 * currently queued packets with fragments that were already sent. So
	 * these
	 * packets won't fail because of the wait time!
	 * 
	 * @param waitTime
	 *            the time that was waited...
	 */
	private void updatePacketTime(final long waitTime) {
		// XXX: not implemented!
	}
}
